package inc.gome.dynamic.datasource.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.net.InetSocketAddress;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

public class ClientFactory {

    int poolSize = 1;

    //客户端工作
    NioEventLoopGroup clientWorker;
    Random rand = new Random();



    private ClientFactory() {
    }

    private static final ClientFactory factory;

    static {
        factory = new ClientFactory();
    }

    public static ClientFactory getFactory() {
        return factory;
    }


    //一个consumer 可以连接很多的provider，每一个provider都有自己的pool  K,V
    ConcurrentHashMap<InetSocketAddress, ClientPool> outboxs = new ConcurrentHashMap<>();


    /**
     * 实现完成回调功能
     *
     * @param requestBody 客户端请求
     * @return
     */
    public static CompletableFuture<Object> transport(RequestBody requestBody) {
        CompletableFuture<Object> res = new CompletableFuture<>();

        try {
            ByteArrayOutputStream out = new ByteArrayOutputStream();
            ObjectOutputStream oout = new ObjectOutputStream(out);
            oout.writeObject(requestBody); //写入对象
            byte[] bytes = out.toByteArray();  //获取对象字节
            //获取一个连接客户端
            NioSocketChannel client = factory.getClient(new InetSocketAddress("localhost", 9090));
            //传输数据
            ByteBuf byteBuf = PooledByteBufAllocator.DEFAULT.directBuffer(bytes.length);
            //客户端注入回调
            ResponseMappingCallback.addCallBack(requestBody.getRequestId(), res);
            byteBuf.writeBytes(bytes);
            System.out.println("============>"+byteBuf.readableBytes());
            //发送服务端
            client.writeAndFlush(byteBuf);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return res;

    }


    //获取客户端连接
    public synchronized NioSocketChannel getClient(InetSocketAddress address) {

        System.out.println("连接池key====>"+address);

        ClientPool clientPool = outboxs.get(address);

        System.out.println("获取到的连接池====>" + clientPool);
        //map
        if (clientPool == null) {
            outboxs.putIfAbsent(address, new ClientPool(poolSize));
            clientPool = outboxs.get(address);
        }

        int i = rand.nextInt(poolSize);

        //获取连接
        if (clientPool.clients[i] != null && clientPool.clients[i].isActive()) {
            NioSocketChannel client = clientPool.clients[i];
            System.out.println("client  ========> status:"+client.isActive());
            return client ;
        }


        synchronized (clientPool.lock[i]) {
            return clientPool.clients[i] = create(address);
        }

    }


    //创建连接
    private NioSocketChannel create(InetSocketAddress address) {

        //基于 netty 的客户端创建方式
        clientWorker = new NioEventLoopGroup(5);
        Bootstrap bs = new Bootstrap();
        ChannelFuture connect = bs.group(clientWorker)
                .channel(NioSocketChannel.class)
                .option(
                        ChannelOption.TCP_NODELAY, true).option(ChannelOption.SO_KEEPALIVE, true).option(
                        ChannelOption.CONNECT_TIMEOUT_MILLIS, 6000).option(
                        ChannelOption.SO_SNDBUF,10248).option(ChannelOption.SO_RCVBUF,
                        12048).handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ChannelPipeline p = ch.pipeline();
                        p.addLast(new ClientDecode());
                        p.addLast(new ByteArrayToBinaryEncoder());
                        p.addLast(new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                //客户端接收响应 服务端返回
                                ResponseBody responseBody = (ResponseBody) msg;

                                //触发回调
                                ResponseMappingCallback.runCallBack(responseBody);


                                System.out.println("接收响应===>" + responseBody.getRequestId());
                            }
                        });

                    }
                }).connect(address);
        try {
            NioSocketChannel client = (NioSocketChannel) connect.sync().channel();
            return client;
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return null;


    }


}
